package com.itlong.cloud.commons.thread.queue.process.consumer;

import com.alibaba.fastjson.JSONObject;
import com.itlong.cloud.POJO.DTO.property.CloudElevatorIotCallCtrlMessageDTO;
import com.itlong.cloud.POJO.DTO.property.CloudElevatorIotCallCtrlV10XMessageDTO;
import com.itlong.cloud.POJO.PO.SendCallElevatorMessageTaskPO;
import com.itlong.cloud.app.dao.IAppDeviceDao;
import com.itlong.cloud.commons.thread.queue.process.queue.SendIotCallElevatorQueue;
import com.itlong.cloud.commons.thread.queue.process.receiver.SendCallElevatorMessageReceiver;
import com.itlong.cloud.constants.RabbitMQConstant;
import com.itlong.cloud.constants.RedisConstant;
import com.itlong.cloud.enumerate.CloudElevatorIotOrderTypeEnum;
import com.itlong.cloud.enumerate.DeviceIotEnum;
import com.itlong.cloud.enumerate.IotCloudCtrlInstructEnum;
import com.itlong.cloud.rabbitmq.config.MessageSender;
import com.itlong.cloud.service.RedisService;
import com.itlong.cloud.utils.node.JsonUtil;
import com.itlong.cloud.utils.words.StringHandlerUtil;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <desc>
 *      发送呼梯消息至iot中心自定义队列消费者
 * </desc>
 *
 * @createDate 2018/04/10
 */
@Service
public class SendCallElevatorMessageConsumer {
    private static final Logger LOG = LoggerFactory.getLogger(SendCallElevatorMessageConsumer.class);
    @Autowired
    SendIotCallElevatorQueue sendIotCallElevatorQueue;
    @Autowired
    MessageSender messageSender;
    private ExecutorService cachedThreadPool ; //线程池
    @Autowired
    RedisService redisService;
    @Autowired
    IAppDeviceDao iAppDeviceDao;
    /**
     * <desc>
     *      线程开始轮询
     * </desc>
     *
     * @param workerId
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    public void worker(int workerId) {
        SendCallElevatorMessageReceiver receiver = new SendCallElevatorMessageReceiver(workerId, sendIotCallElevatorQueue.getSendIotCallElevatorQueue());
        while (true) {
            try {
                //读取队列中的消息
                List<SendCallElevatorMessageTaskPO> sends = receiver.receiveMessage();
                Map<String,Object> productKey = new HashMap<>();        //云电梯产品key
                Map<String,Object> sourceFroms = new HashMap<>();      //命令发起方标识map集合 key为deviceUnique
                Map<String,Object> cmdVers = new HashMap<>();           //命令版本集合 key为deviceUnique
                Map<String,Object> outDatas = new HashMap<>();              //外呼指令集合 key为deviceUnique
                Map<String,Object> inDatas = new HashMap<>();               //内招指令集合 key为deviceUnique
                Map<String,Object> version = new HashMap<>();           //设备固件版本
                Set<String> deviceUniques = new HashSet<>();            //需要推送的设备集合 用于去重
                sends.forEach(send->{
                    //如果需要推送的设备第一次出现
                    if(deviceUniques.add(send.getDeviceUnique())){
                        sourceFroms.put(send.getDeviceUnique(),send.getSourceFrom());
                        cmdVers.put(send.getDeviceUnique(),send.getCmdVer());
                        productKey.put(send.getDeviceUnique(),send.getIotProductKey());
                        version.put(send.getDeviceUnique(),send.getVersion());
                        List<Map<String,Object>> outList = new ArrayList<>();
                        if(send.getOutData()!=null) {
                            outList.add(send.getOutData());
                            outDatas.put(send.getDeviceUnique(),outList);
                        }
                        List<Map<String,Object>> inList = new ArrayList<>();
                        if(send.getInData()!=null){
                            inList.add(send.getInData());
                            inDatas.put(send.getDeviceUnique(),inList);
                        }
                    }else{
                        List<Map<String,Object>> outList = (List<Map<String,Object>>)outDatas.get(send.getDeviceUnique());
                        List<Map<String,Object>> inList = (List<Map<String,Object>>)inDatas.get(send.getDeviceUnique());
                        if(send.getOutData()!=null) {
                            outList.add(send.getOutData());
                            outDatas.put(send.getDeviceUnique(), outList);
                        }
                        if(send.getInData()!=null){
                            inList.add(send.getInData());
                            inDatas.put(send.getDeviceUnique(),inList);
                        }
                    }
                });
                List<Map<String, Object>> controls = new ArrayList<>();
                for(String deviceUnique : deviceUniques){
                    Integer deviceType = StringHandlerUtil.getDeviceTypeByDeviceUnique(deviceUnique);
                    String v = version.get(deviceUnique).toString();
                    if(StringUtils.isBlank(v) || Integer.valueOf(v.substring(v.indexOf("V") + 1, v.indexOf("V") + 5)) < 1040) {
                        //遍历需要发布控制指令的电梯并发布消息
                        CloudElevatorIotCallCtrlMessageDTO message = new CloudElevatorIotCallCtrlMessageDTO();
                        message.setSourceFrom(sourceFroms.get(deviceUnique).toString());
                        //外呼指令
                        if (outDatas.get(deviceUnique) != null && ((List<Map<String, Object>>) outDatas.get(deviceUnique)).size() > 0) {
                            Map<String, Object> out = new HashMap<>();
                            out.put("cmdVer", "0");                  //命令版本 默认为0
                            out.put("data", outDatas.get(deviceUnique));
                            message.setOut(out);
                        }
                        //内呼指令
                        if (inDatas.get(deviceUnique) != null && ((List<Map<String, Object>>) inDatas.get(deviceUnique)).size() > 0) {
                            Map<String, Object> in = new HashMap<>();
                            in.put("cmdVer", "0");
                            in.put("data", inDatas.get(deviceUnique));
                            message.setIn(in);
                        }
                        //发送电梯控制指令
                        Map<String, Object> control = new HashMap<>();
                        control.put("productKey", productKey.get(deviceUnique));
                        control.put("message", JsonUtil.toJSON(message));
                        control.put("topic", "/" + (deviceType == 5 ? deviceUnique + "/user" : deviceUnique) + IotCloudCtrlInstructEnum.DEVICECTRL.getUrl());
                        controls.add(control);
                    }else{
                        CloudElevatorIotCallCtrlV10XMessageDTO message = new CloudElevatorIotCallCtrlV10XMessageDTO();
                        message.setCmd(CloudElevatorIotOrderTypeEnum.FIXEDAMOUNT.getType());//1003
                        message.setSourceFrom(sourceFroms.get(deviceUnique).toString());
                        message.setSnr(0L);
                        message.setTimestamp((new Date()).getTime()+"");
                        Map<String,Object> messageDataChild = new JSONObject();
                        //外呼指令
                        if (outDatas.get(deviceUnique) != null && ((List<Map<String, Object>>) outDatas.get(deviceUnique)).size() > 0) {
                            Map<String, Object> out = new HashMap<>();
                            out.put("cmdVer", "0");                  //命令版本 默认为0
                            out.put("data", outDatas.get(deviceUnique));
                            messageDataChild.put("out", out);
                        }
                        //内呼指令
                        if (inDatas.get(deviceUnique) != null && ((List<Map<String, Object>>) inDatas.get(deviceUnique)).size() > 0) {
                            Map<String, Object> in = new HashMap<>();
                            in.put("cmdVer", "0");
                            in.put("data", inDatas.get(deviceUnique));
                            messageDataChild.put("in", in);
                        }
                        message.setData(messageDataChild);

                        //发送电梯控制指令
                        Integer sendVersion = null;
                        if(deviceType == 1) {
                            String sv = StringHandlerUtil.getNullOrStringForObject(redisService.get(RedisConstant.ELEVATOR_PRODUCT_VERSION_REDIS_KEY));
                            if(StringUtils.isNotBlank(sv)){
                                sendVersion = Integer.valueOf(sv);
                            }else{
                                sendVersion = iAppDeviceDao.getElevatorProductVersion();
                                redisService.set(RedisConstant.ELEVATOR_PRODUCT_VERSION_REDIS_KEY, RedisConstant.ELEVATOR_PRODUCT_VERSION_CACHE_TIME,sendVersion);
                            }
                        }
                        Map<String, Object> control = new HashMap<>();
                        control.put("productKey", productKey.get(deviceUnique));
                        control.put("message", JsonUtil.toJSON(message));
                        control.put("topic", "/" + (deviceType == 5 || (sendVersion!=null && sendVersion == 1) ? deviceUnique + "/user" : deviceUnique) + IotCloudCtrlInstructEnum.DEVICECMD.getUrl());
                        controls.add(control);
                    }
                }
                Map<String,Object> params = new HashMap<>();
                params.put("deviceOperateIotType", DeviceIotEnum.PUB_MESSAGE.getCommandType());
                params.put("devices",controls);
                params.put("timestamp", (new Date()).getTime());
                JSONObject obj = new JSONObject();
                obj.put("data",params);
                messageSender.send(RabbitMQConstant.RABBITMQ_IOT_PUB_QUEUE, obj.toJSONString(),
                        RabbitMQConstant.RABBITMQ_IOT_PUB_EXCHANGE, RabbitMQConstant.RABBITMQ_IOT_PUB_KEY);
            } catch (Exception e) {
                LOG.error("【发送呼梯消息数据异常】",e);
            }
        }
    }

    /**
     * <desc>
     *      初始化
     * </desc>
     *
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    @PostConstruct
    public void polling(){
        //初始化线程池
        cachedThreadPool = Executors.newFixedThreadPool(10);
        //开启10个线程消费消息
        for(int i=1;i<=10;i++){
            int id=i;
            cachedThreadPool.execute(() -> {
                worker(id);
            });
        }
    }
}
